package hbase

import (
	"EasyData/global"
	"fmt"
	"EasyData/util"
	"strings"
	"errors"
)
/**
	将 Hbase的数据格式转化为Mybsql数据的格式
	hbase表名，如果没配置，默认用mysql的表名
	rowKey，如果没配置，默认就是表的id，如果有配置，则是按照配置来进行。
	如果没有配置列族，则将所有的字段全部加到 名为d 的列族中
	mysql字段均采用小写
 */
const defaultFamilyKey  ="d"

func sqlToBase(item map[string]interface{}, task *global.Task) (values map[string]map[string][]byte,table string,key string,errs error) {

	tableName := task.HbaseTable
	if tableName =="" {
		tableName =task.Table
	}

	var valueResult map[string]map[string][]byte
	var err error =nil
	var rowKey string

	if len(task.CloumnFamily)==0 {
		valueResult,err = getDefaultColumn(item,task)
	}else{
		valueResult,err = getByTaskConfig(item,task)
	}
	rowKey,err = buildRowKey(item,task)
	return valueResult,tableName,rowKey,err
}

/**
	以默认的列族组织数据
 */
func getDefaultColumn(item map[string]interface{},task *global.Task) (map[string]map[string][]byte,error){

	var family = make(map[string]map[string][]byte)
	var cells = make(map[string][]byte)
	for i := range  item {
		var a interface{}
		a = item[i]
		res,err :=util.EncodeToByte(a)
		if err !=nil {
			return nil,err
		}
		cells[getKey(i,task)]=res
	}
	//删除id列
	delete(cells,getKey(task.Key,task))
	//删除同步所用的状态字段的列
	if ""!= task.SyncColumn {
		delete(cells,getKey(task.SyncColumn,task))
	}
	family[defaultFamilyKey]=cells
	return family,nil
}
/**
	以配置的列族组织数据
 */
func getByTaskConfig(item map[string]interface{}, task *global.Task) (map[string]map[string][]byte, error) {
	//将mysql的数据列全部转化为小写，好比较
	var mysqls =make(map[string]interface{},len(item))
	for i := range item {
		//排除主键 和 同步数据用到的状态列的字段（当导切数据模式为 仅导入数据到Hbase,不删除mysql的数据时，SyncColumn列会被用到 ）
		if task.Key != i && task.SyncColumn !=i {
			mysqls[strings.ToLower(i)] = item[i]
		}
	}
	taskKey :=  task.CloumnFamily
	var family = make(map[string]map[string][]byte)
	for v := range taskKey{
		familyKey :=taskKey[v]
		for j := range familyKey{
			var cells = make(map[string][]byte)
			columns := familyKey[j]
			for k := range columns{
				//如果在列族里配置里mysql数据库中并不存在的列，则抛出错误
				if _, ok := mysqls[strings.ToLower(columns[k])]; ok {
				}else{
					errStr:=fmt.Sprintf("your hbase column_family have wrong cloumn [%s] which mysql table [%s] cant't find this cloumn",
						strings.ToLower(columns[k]),task.Table)
					return nil,errors.New(errStr)
				}
				res,err :=util.EncodeToByte(mysqls[strings.ToLower(columns[k])])
				if err !=nil {
					return nil,err
				}
				cells[getKey(columns[k],task)]=res
				//同时删除mysqls map中对应的key
				delete(mysqls,strings.ToLower(columns[k]))
			}
			family[j]=cells
		}
	}
	//如果mysqls的数据列仍然有没有迁移到Hbase中的列，同时在config配置中，也没有开启task.PartOf，即不允许只导入部分列到Hbase中，则抛出异常！
	if len(mysqls)!=0 && !task.PartOf {
		errStr := fmt.Sprintf("task.PartOf is [%t] ,but you have some keys not into hbase [%s]",
			task.PartOf,mysqls)
		return nil,errors.New(errStr)
	}
	return family,nil
}

/**
	列存储到Hbase时，大小写转换的问题
	根据配置项决定
 */
func getKey(src string, task *global.Task) (key string) {
	if task.ToUpper{
		return strings.ToUpper(src)
	}
	if task.ToLower {
		return strings.ToLower(src)
	}
	return src
}

/**
	rowKey生成器，进行rowKey的创建
	rowKey中查询到的字段不能为null
 */
func buildRowKey(rows map[string]interface{}, task *global.Task) (string,error) {

	if ""==task.RowKey {
		return  fmt.Sprintf("%v",rows[task.Key]),nil
	}
	//将mysql的数据列全部转化为小写，好比较
	var mysqls =make(map[string]interface{},len(rows))
	for i := range rows {
		mysqls[strings.ToLower(i)] = rows[i]
	}

	//rowKey的每一个key的附加条件列表
	keyParamsMap := make(map[string]*global.KeyParameter)
	for v:= range task.KeyParam {
		param := task.KeyParam[v]
		keyParamsMap[strings.ToLower(param.Key)]=param
	}

	rowKeyStr := ""
	rowKeys := strings.Split(task.RowKey,",")
	for i := range rowKeys{
		column := strings.ToLower(rowKeys[i])
		value := util.ToString(mysqls[column])
		if value== "" {
			errStr := fmt.Sprintf("rowKey cloumn[%s] value is empty",rowKeys[i])
			return "",errors.New(errStr)
		}
		//根据rowKey的不同的条件字段处理
		if _, ok := keyParamsMap[column]; ok {
			param := keyParamsMap[column]

			//如果parameter大于0，则表示配置了长度，则如果长度本身的字段不够，则需要在左侧位补0
			if param.Length>0 {
				value=util.SpaceStr(value,param.Length)
			}
			//如果字段需要反转，则进行反转
			if param.IsReverse {
				value =util.ReverseStr(value)
			}
			//如果字段是时间字段，则去除- : 空格等符号
			if param.IsTime{
				value = strings.Replace(value,"-","",-1)
				value = strings.Replace(value,":","",-1)
				value = strings.Replace(value," ","",-1)
			}
		}
		rowKeyStr =rowKeyStr+value+task.Separator
	}
	rowKeyStr = rowKeyStr[0:len(rowKeyStr)-1]
	return  rowKeyStr,nil
}